package com.learm.test;

import java.util.Arrays;
import java.util.Iterator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

public class WordCount {
	public static void main(String[] args) {
		System.setProperty("hadoop.home.dir", "D:\\SystemEnvironment\\hadoop-2.7.3");
		
		/**
		 *
local 本地单线程
local[K] 本地多线程（指定K个内核）
local[*] 本地多线程（指定所有可用内核）
spark://HOST:PORT 连接到指定的 Spark standalone cluster master，需要指定端口。
mesos://HOST:PORT  连接到指定的  Mesos 集群，需要指定端口。
yarn-client客户端模式 连接到 YARN 集群。需要配置 HADOOP_CONF_DIR。
yarn-cluster集群模式 连接到 YARN 集群。需要配置 HADOOP_CONF_DIR。
		 * */
		
		long count = 0l;
		
		SparkConf conf = new SparkConf()
				.setMaster("local[4]") //开启一个4线程的
				.setAppName("wc");
		JavaSparkContext sc =new JavaSparkContext(conf);

		JavaRDD<String> text = sc.textFile("E:\\workspace\\IdeaProjects\\spark_learn\\LICENSE");
		
		System.out.println("-----------------------------------------------------------------------------------------");
		Long l = System.currentTimeMillis();
		System.out.println(l);

		JavaRDD<String> words = text.flatMap(new FlatMapFunction<String, String>(){
			private static final long serialVersionUID = 1L;
			@Override
			public Iterator<String> call(String line) throws Exception {
				return Arrays.asList(line.split(" ")).iterator();
			}
		});

		JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>(){
			private static final long serialVersionUID = 1L;
			@Override
			public Tuple2<String, Integer> call(String word) throws Exception {
				return new Tuple2<String, Integer>(word, 1);
			}
		});
		JavaPairRDD<String, Integer> results = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Integer call(Integer value1, Integer value2) throws Exception {
				return value1 + value2;
			}
		});
		JavaPairRDD<Integer, String> temp = results.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {
				return new Tuple2<Integer, String>(tuple._2, tuple._1);
			}
		});
        System.out.println("-----------------------------------------------------------------------------------------");
        JavaPairRDD<String, Integer>  sorted = temp.sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple) throws Exception {
				return new Tuple2<String, Integer>(tuple._2, tuple._1);
			}
		});
        System.out.println("-----------------------------------------------------------------------------------------");
        count = sorted.count();
        sorted.foreach(new VoidFunction<Tuple2<String,Integer>>() {
			private static final long serialVersionUID = 1L;
			@Override
			public void call(Tuple2<String, Integer> tuple) throws Exception {
				System.out.println("word: " + tuple._1 + " count: " + tuple._2);
			}
		});

		System.out.println(count);
		System.out.println("-----------------------------------------------------------------------------------------");
		Long l2 = System.currentTimeMillis();
		System.out.println(l2);

		System.out.println(l2 - l);
		sc.close();
	}
}
